package com.example.demo.mqserver.datacenter;

import com.example.demo.common.BinaryTool;
import com.example.demo.common.MqException;
import com.example.demo.mqserver.core.Binding;
import com.example.demo.mqserver.core.Exchange;
import com.example.demo.mqserver.core.MSGQueue;
import com.example.demo.mqserver.core.Message;
import org.springframework.ui.ConcurrentModel;

import java.io.IOException;
import java.security.PublicKey;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: 张
 * Date: 2024-08-13
 * Time: 10:45
 */

/**
 * 该类后续使用的场景可能会涉及到线程安全问题,所以使用ConcurrentHashMap代替HashMap
 */
//管理内存中的数据
public class MemoryDataCenter {
    // key 是 ExchangeName . value 是 Exchange对象   -----   由于肯定会涉及到线程安全问题,此处使用ConcurrentHashMap
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // key 是 queueName,value是MSGQueue对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 第一个Key为ExchangeName,第二个Key为QueueName,value为绑定关系
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // key 是 messageId,value 为 message对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // 表示消息和队列的从属关系,Key 为 queueName,value 为 装载消息的LinkedList
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // 表示待确认的消息 第一个Key 为 queueName,第二个 Key 为 MessageId , value 为 Message对象
    private ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> queueMessageMapWaitAck = new ConcurrentHashMap<>();

    //针对交换机相关API
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(),exchange);
        System.out.println("[MemoryDataCenter] 新交换机创建成功," +
                "exchangeName = " + exchange.getName());
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机移除成功," +
                "exchangeName = " + exchangeName);
    }

    //针对队列相关API
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(),queue);
        System.out.println("[MemoryDataCenter] 队列新建成功," +
                "queueName = " + queue.getName());
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功," +
                "queueName = " + queueName );
    }

    //针对绑定
    public void insertBinding(Binding binding) throws MqException {
        //先使用exchangeName 查一下,对应的hash表是否存在,不存在就先创建一个
//        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
//        if (bindingMap == null) {
//            bindingMap = new ConcurrentHashMap<>();
//            bindingsMap.put(binding.getEx changeName(), bindingMap);
//        }
        //这一行代码效果等于上面的代码 --- 单纯一个是线程安全的
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());
        synchronized (bindingMap) {
            // 再根据queueName 查找,如果已经存在就抛出异常,不存在再进行插入
            if(bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在, exchangeName = " + binding.getExchangeName() +
                        " , queueName = " + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(),binding);
            //上面涉及到了get和put俩步操作,存在线程不安全问题(两个操作运行前因后果)
            System.out.println("[MemoryDataCenter] 添加绑定成功, exchangeName = " + binding.getExchangeName()
                    + " ,queueName = " + binding.getQueueName());
        }
    }

    //获取绑定写两个版本
    //1.根据 exchangeName 和 queueName 确定唯一一个Binding
    public Binding getBinding(String exchangeName,String queueName) {
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap == null) {
            return null;//该交换机没有进行任何绑定
        }
        return bindingMap.get(queueName);
    }
    //2,根据 exchangeName 获取所有的Binding
    public ConcurrentHashMap<String,Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    //删除绑定
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if(bindingMap == null) {
            //该交换机没有进行任何绑定
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName = " + binding.getExchangeName() +
                    ", queueName = " + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 解除绑定成功,exchangeName = " + binding.getExchangeName() +
                " ,queueName = " + binding.getQueueName());
    }

    // 添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 新消息添加成功," +
                "MessageId = " + message.getMessageId());
    }

    //根据id查询消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    //根据Id删除消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息被移除," +
                "MessageId = " + messageId);
    }

    //发送消息到指定队列
    public void sendMessage(MSGQueue queue,Message message) {
        //把消息放到对应的队列中

//        LinkedList<Message> messages = queueMessageMap.get(queue.getName());
//        if(message == null) {
//            messages = new LinkedList<>();
//            queueMessageMap.put(queue.getName(),messages);
//        }
        //1.先根据队列名称找到对应队列的消息链表
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(),k -> new LinkedList<>()); //一步到胃
        //2.把数据添加到messages里面,存在线程安全问题(插入的位置可能会被插队)
        synchronized (messages) {
            messages.add(message);
        }
        addMessage(message);//往消息中心插入一下
        System.out.println("[MemoryDataCenter] 消息被投递到队列中,messageId = " +
                message.getMessageId());
    }

    //从队列中取出消息
    public Message pollMessage(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null) {
            return null;
        }
        synchronized (messages) {
            if(messages.size() == 0) {
                return null;//这里存在线程安全问题
            }
            //链表中有元素就从头取(头删)
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出!! messageId = " + currentMessage.getMessageId());
            return currentMessage;
        }
    }

    //获取指定队列中消息的个数
    public Integer getMessageCount(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null) {
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }

    //添加未确认消息
    public void addMessageWaitAck(String queueName,Message message) {
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageMapWaitAck.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列!! messageId = " + message.getMessageId());
    }

    //删除未确认消息
    public void removeMessageWaitAck(String queueName,String messageId) {
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageMapWaitAck.get(queueName);
        if(messageHashMap == null) {
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列删除!! messageId = " + messageId);
    }

    //获取指定的未确认消息
    public Message getMessageWaitAcK(String queueName,String messageId) {
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageMapWaitAck.get(queueName);
        if(messageHashMap == null) {
            return null;
        }
        return messageHashMap.get(messageId);
    }

    //从硬盘中重新读取数据(恢复数据)
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        // 0.清空之前所有的数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        // 1.恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(),exchange);
        }
        // 2.恢复所有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue :queues) {
            queueMap.put(queue.getName(),queue);
        }
        // 3.恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String,Binding> bindingConcurrentHashMap= bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingConcurrentHashMap.put(binding.getQueueName(),binding);
        }
        // 4.恢复所有的消息
        //   遍历所有的队列,在根据每个队列名字来去获取到所有的消息
        for (MSGQueue queue : queues) {
            //读取该队列所有的消息数据
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueues(queue.getName());
            queueMessageMap.put(queue.getName(),messages);
            for (Message message : messages) {
                messageMap.put(message.getMessageId(),message);
            }
        }
        //注: 针对"未确认消息"这部分内存中的数据,不需要从硬盘中恢复,在设计的时候也没考虑到这一块
        //一旦在等待Ack的过程中,服务器重启了,"未被确认的消息"就会变为"未被取走的消息",这个消息在硬盘上存储的时候就是被当作未被取走的消息进行存储的
    }



}

